package com.province.platform.core.job;

import com.alibaba.fastjson2.JSON;
import com.province.platform.common.enums.CommonErrorCode;
import com.province.platform.common.enums.ContainerType;
import com.province.platform.common.enums.PluginType;
import com.province.platform.common.exception.PlatformException;
import com.province.platform.common.spi.Reader;
import com.province.platform.common.spi.Writer;
import com.province.platform.common.utils.Configuration;
import com.province.platform.core.AbstractContainer;
import com.province.platform.core.assign.AbstractAssign;
import com.province.platform.core.assign.RandomAssign;
import com.province.platform.core.constants.CoreConstants;
import com.province.platform.core.classLoader.ClassLoaderSwapper;
import com.province.platform.core.job.scheduler.AbstractScheduler;
import com.province.platform.core.utils.LoadUtils;
import org.apache.commons.lang3.Validate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;

public class JobContainer extends AbstractContainer {
    private static final Logger LOG = LoggerFactory.getLogger(JobContainer.class);

    private long jobId;

    private long startTimeStamp;

    private long endTimeStamp;

    private String readerPluginName;

    private String writerPluginName;

    private int channelNumber;

    private int totalStage = 1;

    private Reader.Job jobReader;

    private Writer.Job jobWriter;

    private ClassLoaderSwapper classLoaderSwapper = ClassLoaderSwapper.newInstance();

    public JobContainer(Configuration configuration) {
        super(configuration);
    }

    @Override
    public void start() {
        LOG.info("Platform jobContainer starts job.");

        this.startTimeStamp = System.currentTimeMillis();

        LOG.debug("JobContainer starts to do init ...");
        this.init();

        LOG.debug("JobContainer starts to do prepare ...");
        this.prepare();

        LOG.debug("JobContainer starts to do split");
        this.totalStage = this.split();

        LOG.info("JobContainer starts to do schedule");
        this.schedule();
    }

    private void init() {
        this.jobId = this.configuration.getLong(CoreConstants.CORE_CONTAINER_JOB_ID, -1);

        if (this.jobId < 0) {
            LOG.info("Set jobId to zero");
            this.jobId = 0;
            this.configuration.set(CoreConstants.CORE_CONTAINER_JOB_ID, this.jobId);
        }

        Thread.currentThread().setName("Job-" + this.jobId);

        this.jobReader = this.initJobReader();
        this.jobWriter = this.initJobWriter();
    }

    private Reader.Job initJobReader() {
        this.readerPluginName = this.configuration.getString(CoreConstants.JOB_CONTENT_READER_NAME);
        classLoaderSwapper.setClassLoader(LoadUtils.getJarLoader(PluginType.READER, this.readerPluginName));

        Reader.Job jobReader = (Reader.Job) LoadUtils.loadPlugin(PluginType.READER, this.readerPluginName, ContainerType.Job);

        jobReader.init();

        classLoaderSwapper.restoreClassLoader();
        return jobReader;
    }

    private Writer.Job initJobWriter() {
        this.writerPluginName = this.configuration.getString(CoreConstants.JOB_CONTENT_WRITER_NAME);
        classLoaderSwapper.setClassLoader(LoadUtils.getJarLoader(PluginType.WRITER, this.writerPluginName));

        Writer.Job jobWriter = (Writer.Job) LoadUtils.loadPlugin(PluginType.WRITER, this.writerPluginName, ContainerType.Job);

        jobWriter.init();

        classLoaderSwapper.restoreClassLoader();
        return jobWriter;
    }

    private void prepare() {
        this.prepareJobReader();
        this.prepareJobWriter();
    }

    private void prepareJobReader() {
        classLoaderSwapper.setClassLoader(LoadUtils.getJarLoader(PluginType.READER, this.readerPluginName));

        LOG.info("Reader.Job [{}] do prepare work.", this.readerPluginName);
        this.jobReader.prepare();

        classLoaderSwapper.restoreClassLoader();
    }

    private void prepareJobWriter() {
        classLoaderSwapper.setClassLoader(LoadUtils.getJarLoader(PluginType.WRITER, this.writerPluginName));

        LOG.info("Writer.Job [{}] do prepare work.", this.writerPluginName);
        this.jobWriter.prepare();

        classLoaderSwapper.restoreClassLoader();
    }


    private int split() {
//        this.readerChannel = this.configuration.getInt(CoreConstants.JOB_CONTENT_READER_CHANNEL, 1);

        List<Configuration> readerTaskConfigs = this.doReaderSplit();
        int taskNumber = readerTaskConfigs.size();
//        this.writerChannel = this.configuration.getInt(CoreConstants.JOB_CONTENT_WRITER_CHANNEL, taskNumber);
        List<Configuration> writerTaskConfigs = this.doWriterSplit(taskNumber);

        List<Configuration> contentConfig = this.mergeReaderAndWriterTaskConfigs(readerTaskConfigs, writerTaskConfigs);

        LOG.debug("contentConfig configuration: "+ JSON.toJSONString(contentConfig));

        this.configuration.set(CoreConstants.JOB_CONTENT, contentConfig);

        return contentConfig.size();
    }

    private List<Configuration> mergeReaderAndWriterTaskConfigs(List<Configuration> readerTaskConfigs, List<Configuration> writerTaskConfigs) {
        if (readerTaskConfigs.size() != writerTaskConfigs.size()) {
            throw PlatformException.asPlatformException(CommonErrorCode.PLUGIN_SPLIT_ERROR, String.format("reader切分的task数目[%d]不等于writer切分的task数目[%d].",
                    readerTaskConfigs.size(), writerTaskConfigs.size()));
        }

        List<Configuration> contentConfigs = new ArrayList<>();
        for (int i = 0; i < readerTaskConfigs.size(); i++) {
            Configuration taskConfig = Configuration.newDefault();
            taskConfig.set(CoreConstants.READER_NAME, this.readerPluginName);
            taskConfig.set(CoreConstants.READER_PARAMETER, readerTaskConfigs.get(i));
            taskConfig.set(CoreConstants.WRITER_NAME, this.writerPluginName);
            taskConfig.set(CoreConstants.WRITER_PARAMETER, writerTaskConfigs.get(i));

            taskConfig.set(CoreConstants.TASK_ID, i);
            contentConfigs.add(taskConfig);
        }

        return contentConfigs;
    }

    private List<Configuration> doReaderSplit() {
        classLoaderSwapper.setClassLoader(LoadUtils.getJarLoader(PluginType.READER, this.readerPluginName));

        List<Configuration> configs = this.jobReader.split();

        if (configs == null || configs.size() <= 0) {
            throw PlatformException.asPlatformException(CommonErrorCode.PLUGIN_SPLIT_ERROR, "reader切分的task数目不能小于等于0");
        }

        LOG.info("Reader.Job [{}] splits to [{}] tasks.", this.readerPluginName, configs.size());
        classLoaderSwapper.restoreClassLoader();

        return configs;
    }

    private List<Configuration> doWriterSplit(int adviceNumber) {
        classLoaderSwapper.setClassLoader(LoadUtils.getJarLoader(PluginType.WRITER, this.writerPluginName));

        List<Configuration> configs = this.jobWriter.split(adviceNumber);

        if (configs == null || configs.size() <= 0) {
            throw PlatformException.asPlatformException(CommonErrorCode.PLUGIN_SPLIT_ERROR, "writer切分的task数目不能小于等于0");
        }

        LOG.info("Writer.Job [{}] splits to [{}] tasks.", this.writerPluginName, configs.size());
        classLoaderSwapper.restoreClassLoader();

        return configs;
    }

    private void schedule() {
        Validate.isTrue(Objects.nonNull(this.configuration), "框架获取的job不能为null");
        int channel = this.configuration.getInt(CoreConstants.JOB_SETTING_CHANNEL);
        int taskGroupChannel = this.configuration.getInt(CoreConstants.CORE_CONTAINER_TASK_GROUP_CHANNEL, 5);

        int taskNumber = this.configuration.getList(CoreConstants.JOB_CONTENT).size();
        this.channelNumber = Math.min(channel, taskNumber);
        Validate.isTrue(channelNumber > 0 && taskGroupChannel > 0,
                "每个channel的平均task数[averTaskPerChannel]，channel数目[channelNumber]，每个taskGroup的平均channel数[channelsPerTaskGroup]都应该为正数");

        // 随机分布
        AbstractAssign abstractAssign = new RandomAssign(this.configuration, channelNumber, taskGroupChannel);
        List<Configuration> taskGroupConfigs = abstractAssign.getTaskGroupConfigs();
        LOG.info("Scheduler starts [{}] taskGroups.", taskGroupConfigs.size());

        AbstractScheduler scheduler = null;
    }
}
